package com.bw.yk08;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.TimeUnit;

public class Test05 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> dws_trade_ordersDS=env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dws_trade_orders","test03"));


        SingleOutputStreamOperator<TradeOrdersStats> orderInfoMap=dws_trade_ordersDS.map(new MapFunction<String, TradeOrdersStats>() {
            @Override
            public TradeOrdersStats map(String s) throws Exception {
                OrderInfoWide orderInfoWide=JSON.parseObject(s,OrderInfoWide.class);
                TradeOrdersStats tradeOrdersStats=new TradeOrdersStats();
                if(orderInfoWide!=null)
                {

                    tradeOrdersStats.setProduct_spu_id(orderInfoWide.getProduct_spu_id());
                    tradeOrdersStats.setRegion_code(orderInfoWide.getRegion_code());
                    tradeOrdersStats.setOrder_ct(1L);
                    tradeOrdersStats.setOrder_sku_num(orderInfoWide.getOrderDetail().getSku_num());
                    tradeOrdersStats.setOrder_total_amount(orderInfoWide.getOrderInfo().getActual_amount());
                    tradeOrdersStats.setTs(orderInfoWide.getOrderInfo().getCreate_ts());
                }
                return tradeOrdersStats;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<TradeOrdersStats>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeOrdersStats>() {
                    @Override
                    public long extractTimestamp(TradeOrdersStats tradeOrdersStats, long l) {
                        return tradeOrdersStats.getTs();
                    }
                })
        );

        KeyedStream<TradeOrdersStats, Tuple2<String,Long>> KeyedStreamDS=orderInfoMap.keyBy(new KeySelector<TradeOrdersStats, Tuple2<String,Long>>() {
            @Override
            public Tuple2<String, Long> getKey(TradeOrdersStats tradeOrdersStats) throws Exception {
                return new Tuple2<>(tradeOrdersStats.getRegion_code(),tradeOrdersStats.getProduct_spu_id());
            }
        });

        SingleOutputStreamOperator<TradeOrdersStats> resultDS=KeyedStreamDS
//                .window(TumblingEventTimeWindows.of(Time.days(1)))
//                .trigger(ContinuousEventTimeTrigger.of(Time.seconds(1)))//一秒钟触发一次
                .window(SlidingProcessingTimeWindows.of(Time.days(1),Time.seconds(1)))
//                .sideOutputLateData()
        .reduce(
                new ReduceFunction<TradeOrdersStats>() {
                    @Override
                    public TradeOrdersStats reduce(TradeOrdersStats left, TradeOrdersStats right) throws Exception {
                        left.setOrder_ct(left.getOrder_ct()+right.getOrder_ct());
                        left.setOrder_sku_num(left.getOrder_sku_num()+right.getOrder_sku_num());
                        left.setOrder_total_amount(left.getOrder_total_amount().add(right.getOrder_total_amount()));
                        return left;
                    }
                },
                new WindowFunction<TradeOrdersStats, TradeOrdersStats, Tuple2<String, Long>, TimeWindow>() {

                    @Override
                    public void apply(Tuple2<String, Long> stringLongTuple2, TimeWindow timeWindow, Iterable<TradeOrdersStats> iterable, Collector<TradeOrdersStats> collector) throws Exception {
                        Long start=timeWindow.getStart();
                        Long end=timeWindow.getEnd();

                        SimpleDateFormat sf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                        TradeOrdersStats bean=iterable.iterator().next();
                        bean.setStt(sf.format(start));
                        bean.setEdt(sf.format(end));
                        bean.setTs(new Date().getTime());

                        collector.collect(bean);
                    }
                }
        );



        SingleOutputStreamOperator<TradeOrdersStats> result= AsyncDataStream.unorderedWait(resultDS, new ADSAsyncFunction(), 1000, TimeUnit.MILLISECONDS, 100);

        result.addSink(
                JdbcSink.sink(
                        "insert into yk08_orderwidestate (stt,edt,ts,region_code,product_spu_id,order_ct,order_total_amount,order_sku_num,region_name,spu_name) values (?, ?, ?, ?,?, ?, ?, ?, ?, ?)",
                        (statement, tradeOrdersStats) -> {
                            statement.setString(1, tradeOrdersStats.getStt());
                            statement.setString(2, tradeOrdersStats.getEdt());
                            statement.setLong(3, tradeOrdersStats.getTs());
                            statement.setString(4, tradeOrdersStats.getRegion_code());
                            statement.setLong(5, tradeOrdersStats.getProduct_spu_id());
                            statement.setLong(6, tradeOrdersStats.getOrder_ct());
                            statement.setBigDecimal(7, tradeOrdersStats.getOrder_total_amount());
                            statement.setLong(8, tradeOrdersStats.getOrder_sku_num());
                            statement.setString(9, tradeOrdersStats.getRegion_name());
                            statement.setString(10, tradeOrdersStats.getSpu_name());
                        },
                        JdbcExecutionOptions.builder()
                                .withBatchSize(0)
                                .build(),
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:clickhouse://hadoop-single:8123/default")
                                .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                                .build()

                )
        );

        env.execute();
    }
}
